/*
 * 版权所有 (C) 2015 知启蒙(ZHIQIM) 保留所有权利。[遇见知启蒙，邂逅框架梦]
 * 
 * https://zhiqim.org/project/zhiqim_framework/zhiqim_httpd.htm
 *
 * Zhiqim Httpd is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */
package org.zhiqim.httpd.nio;

import java.io.EOFException;
import java.io.IOException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;

import org.zhiqim.httpd.HttpChannel;
import org.zhiqim.httpd.HttpConnection;
import org.zhiqim.httpd.HttpContext;
import org.zhiqim.httpd.HttpListener;
import org.zhiqim.httpd.websocket.HttpWebSocketService;
import org.zhiqim.kernel.model.maps.HashMapSS;
import org.zhiqim.kernel.util.Closes;
import org.zhiqim.kernel.websocket.WebSocketConstants;

/**
 * HTTP连接信息类，包括输入输出流，实现Runnable，用于线程处理
 *
 * @version v1.0.0 @author zouzhigang 2014-3-21 新建与整理
 */
public class HttpNioConnection implements HttpConnection, HttpChannel, Runnable, WebSocketConstants
{
    private static final LinkedList<Selector> selectorList = new LinkedList<>();
    
    private final HttpListener listener;
    private final SocketChannel socket;
    private final String remoteAddress;
    private final int remotePort;
    
    //连接编号&缓冲&最近消息头
    protected String connId;
    protected ByteBuffer netBuf;
    protected HttpNioHeader header;
    
    //关闭和锁状态
    private volatile boolean closed;
    private volatile boolean locked;
    private volatile long lockTime;
    
    //WebSocket
    protected HttpNioWebSocket websocket;
    
    public HttpNioConnection(HttpListener listener, SocketChannel channel, String remoteAddress, int remotePort)
    {//HTTP
        this.listener = listener;
        this.socket = channel;
        this.remoteAddress = remoteAddress;
        this.remotePort = remotePort;
        
        this.connId = sequence.nextString();
        this.netBuf = ByteBuffer.allocate(listener.getSoRecvBuf());
        
        //初始为解除锁定状态
        this.unlock();
    }
    
    /** 关闭后置，SSL连接可重写 */
    protected void closeAfter()
    {
    }
    
    /** 判断消息头，SSL连接可重写 */
    protected void chkHeader()
    {
        if (header == null || header.isParsed())
            header = new HttpNioHeader(this);
    }
    
    /** 解决缓冲数据，SSL连接可重写 */
    protected int parseBuffer()
    {
        header.addBuffer(netBuf);
        return 1;
    }
    
    /********************************************************************************/
    //连接活动状态
    /********************************************************************************/
    
    /** 尝试取锁 */
    public boolean tryLock()
    {
        synchronized (this)
        {
            if (locked)
            {//活动中，表示被其他线程先执行
                return false;
            }
            
            locked = true;
            lockTime = System.currentTimeMillis();
            return true;
        }
    }
    
    /** 解除锁定 */
    private void unlock()
    {
        synchronized (this)
        {
            locked = false;
            lockTime = System.currentTimeMillis();
        }
    }
    
    /** 是否活动的 */
    public boolean isActive()
    {
        return locked;
    }
    
    /** 是否空闲超时 */
    public boolean isOvertime()
    {
        if (websocket != null || !locked)
            return false;
        
        return (System.currentTimeMillis() - lockTime) > listener.getSoIdleTimeMs();
    }
    
    /** 是否已关闭 */
    public boolean isClosed()
    {
        synchronized (this)
        {
            return closed || !socket.isOpen();
        }
    }
    
    /********************************************************************************/
    //线程消息处理（1、连接中断，2、数据为空，3、有数据，4、连接异常）
    /********************************************************************************/
    
    public void run()
    {
        try
        {
            //3.从通道中读数据，并作相应的处理
            netBuf.clear();
            int count = socket.read(netBuf);
            if (count == -1)
            {//3.1 读数据时客户端已关闭或服务端异常，连接关闭
                closeSelfOrWebsocket();
            }
            else if (count == 0)
            {//3.2 连接数据为空，连接解锁，结束线程
                unlock();
            }
            else
            {//3.3 有连接数据，业务处理
                executeSelfOrWebsocket();
            }
        }
        catch (Throwable e)
        {//3.4 连接异常
            closeSelfOrWebsocketException(e);
        }
    }
    
    /** 有数据处理 */
    private void executeSelfOrWebsocket()
    {
        if (websocket != null)
        {//Websocket消息处理
            int result = websocket.execute(netBuf);
            if (result == -1)
            {//解析失败或异常，关闭
                closeSelfOrWebsocket();
            }
            else
            {//正常或下溢，连接解锁，等待后续数据
                unlock();
            }
        }
        else
        {//HTTP/HTTPS消息处理
            
            //1.检查消息头对象
            chkHeader();
            
            //2.解析缓冲数据
            int result = parseBuffer();
            if (result == -1)
            {//解析失败或异常
                close();
                return;
            }
            else if (result == 0)
            {//下溢，连接解锁，等待后续数据
                unlock();
                return;
            }
            
            try
            {//3.消息处理，然后连接解锁
                header.execute();
                unlock();
            }
            catch (Exception e)
            {//4.失败关闭
                close();
            }
        }
    }
    
    /** 连接中断 */
    private void closeSelfOrWebsocket()
    {
        if (websocket == null){
            close();
        }else{
            websocket.close(CLOSE_SELF);
            websocket = null;
        }
    }
    
    /** 连接异常 */
    private void closeSelfOrWebsocketException(Throwable e)
    {
        if (websocket == null){
            close();
        }else{
            websocket.exception(e);
            websocket = null;
        }
    }
    
    /********************************************************************************/
    //通道读写数据&关闭
    /********************************************************************************/
    
    /** 读取缓冲 */
    public int read(ByteBuffer buf) throws IOException
    {
        return socket.read(buf);
    }
    
    /** 写入缓冲 */
    public int write(ByteBuffer buf) throws IOException
    {
        int count = 0;
        while (buf.hasRemaining())
        {
            int len = socket.write(buf);
            if (len < 0)
            {//客户端主动断开
                throw new EOFException();
            }
            else if (len > 0)
            {//客户端网络正常，写入成功
                count += len;
            }
            else
            {//客户端网络阻塞，启动阻塞检查
                writeSelection();
            }
        }
        
        return count;
    }
    
    /** 写入数据 */
    public void write(byte[] data) throws IOException
    {
        write(data, 0, data.length);
    }
    
    /** 写入数据 */
    public void write(byte[] data, int off, int len) throws IOException
    {
        write(ByteBuffer.wrap(data, off, len));
    }
    
    /** 写入数据阻塞等待 */
    private void writeSelection() throws IOException
    {
        Selector selector = getSelector();
        SelectionKey key = null;
        
        try
        {
            key = socket.register(selector, SelectionKey.OP_WRITE);
            if (selector.select(listener.getSoTimeoutMs()) == 0) 
            {//超时，抛出异常
                throw new SocketTimeoutException();
            } 
        }
        finally
        {
            if (key != null)
            {
                key.cancel();
                key = null;
            }
            
            selector.selectNow();
            setSelector(selector);
        }
    }
    
    /** 关闭连接 */
    public void close()
    {
        synchronized (this)
        {
            if (closed)
                return;
            
            closed = true;
        }

        unlock();
        closeAfter();
        listener.remove(this);
        Closes.closeIgnoreException(socket);
    }
    
    /***********************************************************************/
    //获取连接对应的websocket属性
    /***********************************************************************/
    
    @Override /** 是否是WebSocket连接 */
    public boolean isWebSocket()
    {
        return websocket != null;
    }
    
    @Override /** 升级为WebSocket连接 */
    public void doWebSocket(HttpContext context, HashMapSS headers, HashMapSS parameters, String protocol, String sessionId, HttpWebSocketService service)
    {
        websocket = new HttpNioWebSocket(this, context, headers, parameters, protocol, sessionId, service);
        websocket.open();
    }
    
    @Override /** 开启WebSocket回调 */
    public void doWebSocketOpen()
    {
        if (websocket != null)
        {
            websocket.doOpen();
        }
    }
    
    /***********************************************************************/
    //获取连接属性上的参数
    /***********************************************************************/
    
    @Override
    public String getId()
    {
        return connId;
    }
    
    @Override
    public HttpListener getListener()
    {
        return listener;
    }
    
    @Override
    public String getRemoteAddr()
    {
        return remoteAddress;
    }
    
    @Override
    public int getRemotePort()
    {
        return remotePort;
    }
    
    @Override
    public String toString()
    {
        return new StringBuilder("HttpNioConnection[").append(connId).append("]").toString();
    }
    
    /*************************************************************************/
    //对Selector进行缓存处理
    /*************************************************************************/
    
    private static Selector getSelector() throws IOException
    {
        synchronized (selectorList)
        {
            if (!selectorList.isEmpty())
                return selectorList.removeFirst();
        }
        
        //队列中没有则创建一个
        return Selector.open();
    }
    
    private static void setSelector(Selector selector)
    {
        synchronized (selectorList)
        {
            selectorList.addLast(selector);
        }
    }
}
